package com.mtf.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.mtf.common.enums.IdempotentEnum;
import com.mtf.common.vo.Result;
import com.mtf.entity.MtfOrder;
import com.mtf.entity.MtfRocketmqLog;
import com.mtf.handler.exception.MqRepeatConsumeException;
import com.mtf.mapper.MtfOrderMapper;
import com.mtf.service.IMtfOrderService;
import com.mtf.service.IMtfRocketmqLogService;
import com.mtf.service.IMtfStockService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;

import static com.mtf.common.enums.MsgStatusEnum.CONSUME_SUCCESS;
import static com.mtf.common.enums.MsgStatusEnum.CONSUMING;
import static com.mtf.handler.IdempotentExceptionEnum.MQ_REPEAT_CONSUME;

/**
 * @Description: mtf_order
 * @Author: jeecg-boot
 * @Date:   2023-11-02
 * @Version: V1.0
 */
@Slf4j
@Service
public class MtfOrderServiceImpl extends ServiceImpl<MtfOrderMapper, MtfOrder> implements IMtfOrderService {

    @Autowired
    private IMtfStockService stockService;
    @Autowired
    private IMtfRocketmqLogService rocketmqLogService;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Resource
    private RedisTemplate<String,Object> redisTemplate;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;


    @Override
    public void createOrder(MtfOrder mtfOrder) {
        //先发送半消息
        //执行本地业务
        //库存模块消费消息执行减库存操作
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void insertMtfOrder(MtfOrder mtfOrder) {
        this.baseMapper.insert(mtfOrder);
    }

    /**
     * @Description: 创建订单
     * @param mtfOrder
     * @param idempotentEnum 订单表字段唯一索引、订单表订单查询、消息表唯一索引
     * @return com.mtf.common.vo.Result<?>
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result<?> saveOrder(MtfOrder mtfOrder, IdempotentEnum idempotentEnum) {
        Result<?> result;
        //注意case穿透现象
        switch (idempotentEnum) {
            case SERVICE_IDEMPOTENT:
                //测试该方案：需要为订单表的identity添加唯一索引
                result = saveOrderByServiceIdempotent(mtfOrder);
                break;
            case MESSAGE_TABLE_IDEMPOTENT:
                //测试该方案：需要为消息表的msg_id添加唯一索引（业务逻辑解耦）
                result = saveOrderByMessageTableIdempotent(mtfOrder);
                break;
            case MESSAGE_STATUS_IDEMPOTENT:
                //测试该方案：根据消息表中的消息状态解决重复消费的问题
                result = saveOrderByMessageStatusIdempotent(mtfOrder);
                break;
            case REDIS_IDEMPOTENT:
                //测试该方案：兜底方案，解决第一次消费消息（机器重启等原因）设置超时时间，超过消费时间删除消息记录一律重新消费
                result = saveOrderByRedisIdempotent(mtfOrder);
                break;
            default:
                //测试该方案：删除订单表identity字段的唯一索引（简单的回查判断来解决消息重复消费的问题）
                result = saveServiceSelectIdempotent(mtfOrder);
                break;
        }
        return result;
    }

    /**
     * @Description: 消息表去事务化（根据消息状态解决重复消费问题）
     * @param mtfOrder
     * @return com.mtf.common.vo.Result<?>
     */
    @Transactional(rollbackFor = Exception.class)
    public Result<?> saveOrderByMessageStatusIdempotent(MtfOrder mtfOrder) {
        //判断消息表中是否存在该订单对应的消息及消息状态
        MtfRocketmqLog rocketmqLog = rocketmqLogService.selectRocketMqLogByMsgId(mtfOrder.getIdentity());
        if (ObjectUtil.isNotEmpty(rocketmqLog)&&rocketmqLog.getMsgStatus().equals(CONSUME_SUCCESS.getCode())){
            log.info("订单创建成功...");
            return Result.ok("订单创建成功...");
        }
        if (ObjectUtil.isNotEmpty(rocketmqLog)&&rocketmqLog.getMsgStatus().equals(CONSUMING.getCode())){
            //发送延迟消息
            Message message = new Message();
            message.setTopic("idempotent_topic");
            message.setBody(JSON.toJSONBytes(mtfOrder));
            message.setDelayTimeLevel(4);
            rocketMQTemplate.syncSend("idempotent_topic",message);
            log.info("订单正在创建，请稍后30s后重试...");
            return Result.ok("订单正在创建，请稍后30s后重试...");
        }
        try {
            MtfRocketmqLog mqLog = new MtfRocketmqLog();
            mqLog.setMsgId(mtfOrder.getIdentity());
            mqLog.setSendTime(new Date());
            mqLog.setMessageBody(JSON.toJSONString(mtfOrder));
            mqLog.setBusinessType("创建订单");
            //设置消息的初始状态
            mqLog.setMsgStatus(CONSUMING.getCode());
            rocketmqLogService.save(mqLog);
            //校验库存
            if (!stockService.checkStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum())){
                throw new Exception("库存不足，下单失败！");
                //return Result.error("库存不足，下单失败！");
            }
            //扣减库存
            stockService.substractStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum());
            this.save(mtfOrder);
            //订单创建完成，更新消息状态
            rocketmqLogService.updateRocketMqLogMsgStatusByMsgId(mtfOrder.getIdentity());
        } catch (Exception e) {
            //订单创建失败，删除对应消息，进行重新消费
            rocketmqLogService.deleteRocketMqLogByMsgId(mtfOrder.getIdentity());
        }
        return Result.ok("订单创建成功！");
    }

    /**
     * @Description: 创建订单根据订单表唯一索引解决重复消费问题...
     * 存在的问题：对于分库分表的场景下唯一索引还有效吗？或者怎么优化？
     * @param mtfOrder
     * @return com.mtf.common.vo.Result<?>
     */
    @Transactional(rollbackFor = Exception.class)
    public Result<?> saveOrderByServiceIdempotent(MtfOrder mtfOrder){
        //校验库存
        if (!stockService.checkStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum())){
            return Result.error("库存不足，下单失败！");
        }
        //扣减库存
        stockService.substractStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum());
        try {
            this.save(mtfOrder);
        }
        catch (DuplicateKeyException e) {
            log.info("订单表唯一索引异常:{}",e.getMessage());
            throw new MqRepeatConsumeException(MQ_REPEAT_CONSUME.getDescription()+"：创建订单根据订单表唯一索引解决重复消费问题...");
        }
        //创建订单
        return Result.ok("订单创建成功！");
    }

    /**
     * @Description: 创建订单根据消息表唯一索引解决重复消费问题...
     * 存在的问题：和业务表唯一索引一样，虽然解决重复消费和业务逻辑进行了解耦合，但是都无法跨库处理，
     *          特别业务逻辑中存在不支持事务的数据修改无法回滚的问题而且数据库唯一索引校验会锁表时间过长影响性能
     * @param mtfOrder
     * @return com.mtf.common.vo.Result<?>
     */
    @Transactional(rollbackFor = Exception.class)
    public Result<?> saveOrderByMessageTableIdempotent(MtfOrder mtfOrder){
        //根据消息表标识判断该订单是否已经创建，避免重复消费
        MtfRocketmqLog mtfRocketmqLog = new MtfRocketmqLog();
        mtfRocketmqLog.setMsgId(mtfOrder.getIdentity());
        mtfRocketmqLog.setSendTime(new Date());
        mtfRocketmqLog.setMessageBody(JSON.toJSONString(mtfOrder));
        mtfRocketmqLog.setBusinessType("创建订单");
        try {
            rocketmqLogService.save(mtfRocketmqLog);
        } catch (Exception e) {
            log.info("消息表唯一索引异常:{}",e.getMessage());
            throw new MqRepeatConsumeException(MQ_REPEAT_CONSUME.getDescription()+"：创建订单根据消息表唯一索引解决重复消费问题...");
        }
        //校验库存
        if (!stockService.checkStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum())){
            return Result.error("库存不足，下单失败！");
        }
        //扣减库存
        stockService.substractStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum());
        //创建订单
        this.save(mtfOrder);
        return Result.ok("订单创建成功！");
    }

    public Result<?> saveOrderByRedisIdempotent(MtfOrder mtfOrder){
        //判断消息的状态
        Integer status = (Integer) redisTemplate.opsForHash().get(mtfOrder.getIdentity(), "status");
        if (ObjectUtil.isNotEmpty(status)){
            assert status != null;
            if (status.equals(CONSUMING.getCode())){
                //发送延迟消息到队列
                Message message = new Message();
                message.setTopic("idempotent_topic");
                message.setBody(JSON.toJSONBytes(mtfOrder));
                message.setDelayTimeLevel(4);
                rocketMQTemplate.syncSend("idempotent_topic",message);
                log.info("订单正在创建，请稍后30s后重试...");
                return Result.ok("订单正在创建，请稍后30s后重试...");
            }
            if (status.equals(CONSUME_SUCCESS.getCode())){
                return Result.ok("订单创建成功！");
            }
        }
        MtfRocketmqLog mqLog = new MtfRocketmqLog();
        mqLog.setMsgId(mtfOrder.getIdentity());
        mqLog.setSendTime(new Date());
        mqLog.setMessageBody(JSON.toJSONString(mtfOrder));
        mqLog.setBusinessType("创建订单");
        //设置消息的初始状态
        mqLog.setMsgStatus(CONSUMING.getCode());
        //判断redis缓存中是否存在对应的消息记录以及消息的状态（处于“消费中”的消息过期时间设置为6分钟）
        Map<String, Object> map = new HashMap<>();
        BeanUtils.copyProperties(mqLog,map);
        redisTemplate.opsForHash().putAll(mqLog.getMsgId(),map);
        redisTemplate.expire(mqLog.getMsgId(),360, TimeUnit.SECONDS);
        //校验库存
        if (!stockService.checkStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum())){
            return Result.error("库存不足，下单失败！");
        }
        //扣减库存
        stockService.substractStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum());
        //创建订单
        this.save(mtfOrder);
        //将redis中对应的消息的有效时间设置永不过期
        redisTemplate.persist(mtfOrder.getIdentity());
        return Result.ok("订单创建成功！");
    }

    /**
     * @Description: 创建订单根据订单标识判断订单创建之前是否存在...
     * 存在的问题：高并发场景第一次创建订单还没有落库，第二次进行重复消费会穿透检查的挡板，因此并发去重是一个要考虑的问题
     * @param mtfOrder
     * @return com.mtf.common.vo.Result<?>
     */
    public Result<?> saveServiceSelectIdempotent(MtfOrder mtfOrder){
        //校验当前订单是否存在
        List<MtfOrder> orderList = this.baseMapper.selectOrderByIdentity(mtfOrder.getIdentity());
        if (CollUtil.isNotEmpty(orderList)){
            throw new MqRepeatConsumeException(MQ_REPEAT_CONSUME.getDescription()+"：创建订单根据订单标识判断订单创建之前是否存在...");
        }
        //校验库存
        if (!stockService.checkStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum())){
            return Result.error("库存不足，下单失败！");
        }
        //扣减库存
        stockService.substractStockByProductIdentity(mtfOrder.getProductIdentity(),mtfOrder.getNum());
        //创建订单
        this.save(mtfOrder);
        return Result.ok("订单创建成功！");
    }

    @Override
    public MtfOrder getInfo(String id) {
        return this.baseMapper.getInfo(id);
    }

    @Override
    public MtfOrder getInfos(String id) {
        return this.baseMapper.getInfos(id);
    }
}
